# -*- coding: utf-8 -*-
"""数据比较工具模块
提供数据比较和验证功能，确保数据一致性和准确性
"""

import pandas as pd


class DataComparison:
    """数据比较工具类"""

    @staticmethod
    def compare_records(record1, record2, fields=None):
        """比较两条记录的差异

        Args:
            record1: 第一条记录
            record2: 第二条记录
            fields: 要比较的字段列表，如果为None则比较所有共同字段

        Returns:
            dict: 包含差异字段及其值的字典
        """
        if record1 is None or record2 is None:
            return {}

        # 如果没有指定字段，使用两条记录的共同字段
        if fields is None:
            if isinstance(record1, dict) and isinstance(record2, dict):
                common_fields = set(record1.keys()) & set(record2.keys())
            elif isinstance(record1, pd.Series) and isinstance(
                record2, pd.Series
            ):
                common_fields = set(record1.index) & set(record2.index)
            else:
                return {}
        else:
            common_fields = fields

        differences = {}

        # 比较每个字段
        for field in common_fields:
            value1 = (
                record1.get(field)
                if isinstance(record1, dict)
                else record1.get(field, None)
            )
            value2 = (
                record2.get(field)
                if isinstance(record2, dict)
                else record2.get(field, None)
            )

            # 处理NaN值
            if pd.isna(value1) and pd.isna(value2):
                continue
            elif pd.isna(value1) or pd.isna(value2):
                differences[field] = (value1, value2)
            elif value1 != value2:
                differences[field] = (value1, value2)

        return differences

    @staticmethod
    def compare_dataframes(df1, df2, key_fields=None, ignore_fields=None):
        """比较两个DataFrame的差异

        Args:
            df1: 第一个DataFrame
            df2: 第二个DataFrame
            key_fields: 用于匹配记录的关键字段列表
            ignore_fields: 比较时忽略的字段列表

        Returns:
            dict: 包含新增记录、删除记录和修改记录的差异结果
        """
        if df1 is None or df1.empty:
            return {
                "added": (
                    df2.to_dict("records")
                    if df2 is not None and not df2.empty
                    else []
                ),
                "removed": [],
                "modified": [],
            }

        if df2 is None or df2.empty:
            return {
                "added": [],
                "removed": df1.to_dict("records"),
                "modified": [],
            }

        # 如果没有指定关键字段，使用所有共同字段
        if key_fields is None:
            key_fields = list(set(df1.columns) & set(df2.columns))
            if not key_fields:
                # 如果没有共同字段，无法匹配记录
                return {
                    "added": df2.to_dict("records"),
                    "removed": df1.to_dict("records"),
                    "modified": [],
                }

        # 创建用于比较的字段列表
        compare_fields = list(set(df1.columns) & set(df2.columns))
        if ignore_fields:
            compare_fields = [
                f for f in compare_fields if f not in ignore_fields
            ]

        # 创建记录匹配的键
        def create_key(row):
            return tuple(
                str(row[field]) if not pd.isna(row[field]) else ""
                for field in key_fields
            )

        # 为两个DataFrame创建键到索引的映射
        df1_keys = {create_key(row): idx for idx, row in df1.iterrows()}
        df2_keys = {create_key(row): idx for idx, row in df2.iterrows()}

        # 找出新增和删除的记录
        added_keys = set(df2_keys.keys()) - set(df1_keys.keys())
        removed_keys = set(df1_keys.keys()) - set(df2_keys.keys())

        # 找出修改的记录
        modified_records = []
        common_keys = set(df1_keys.keys()) & set(df2_keys.keys())

        for key in common_keys:
            idx1 = df1_keys[key]
            idx2 = df2_keys[key]
            record1 = df1.iloc[idx1]
            record2 = df2.iloc[idx2]

            differences = DataComparison.compare_records(
                record1, record2, compare_fields
            )
            if differences:
                modified_records.append(
                    {
                        "key": key,
                        "differences": differences,
                        "old_record": record1.to_dict(),
                        "new_record": record2.to_dict(),
                    }
                )

        # 构建结果
        result = {
            "added": [df2.iloc[df2_keys[key]].to_dict() for key in added_keys],
            "removed": [
                df1.iloc[df1_keys[key]].to_dict() for key in removed_keys
            ],
            "modified": modified_records,
        }

        return result

    @staticmethod
    def generate_comparison_report(differences, output_file=None):
        """生成比较报告

        Args:
            differences: 比较结果字典
            output_file: 输出文件路径，如果为None则返回字符串

        Returns:
            str: 比较报告内容
        """
        report = []
        report.append("数据比较报告")
        report.append("=" * 50)

        # 新增记录
        if differences["added"]:
            report.append(f"\n新增记录 ({len(differences['added'])} 条):")
            report.append("-" * 50)
            for i, record in enumerate(differences["added"], 1):
                report.append(f"记录 {i}:")
                for field, value in record.items():
                    report.append(f"  {field}: {value}")
                report.append("")

        # 删除记录
        if differences["removed"]:
            report.append(f"\n删除记录 ({len(differences['removed'])} 条):")
            report.append("-" * 50)
            for i, record in enumerate(differences["removed"], 1):
                report.append(f"记录 {i}:")
                for field, value in record.items():
                    report.append(f"  {field}: {value}")
                report.append("")

        # 修改记录
        if differences["modified"]:
            report.append(f"\n修改记录 ({len(differences['modified'])} 条):")
            report.append("-" * 50)
            for i, mod in enumerate(differences["modified"], 1):
                report.append(f"记录 {i}:")
                report.append("  修改的字段:")
                for field, (old_val, new_val) in mod["differences"].items():
                    report.append(
                        f"    {field}: 从 '{old_val}' 变更为 '{new_val}'"
                    )
                report.append("")

        report_content = "\n".join(report)

        # 如果指定了输出文件，将报告写入文件
        if output_file:
            try:
                with open(output_file, "w", encoding="utf-8") as f:
                    f.write(report_content)
            except Exception as e:
                print(f"写入比较报告失败: {str(e)}")

        return report_content


class DataValidator:
    """数据验证工具类"""

    def __init__(self):
        """初始化数据验证器"""
        self.rules = {}

    def add_validation_rule(self, field_name, rule_func, error_message):
        """添加验证规则

        Args:
            field_name: 字段名
            rule_func: 验证函数，接受字段值作为参数，返回布尔值表示验证是否通过
            error_message: 验证失败时的错误消息
        """
        if field_name not in self.rules:
            self.rules[field_name] = []
        self.rules[field_name].append((rule_func, error_message))

    def validate_record(self, record):
        """验证单条记录

        Args:
            record: 要验证的记录

        Returns:
            list: 验证错误列表，如果为空则表示验证通过
        """
        errors = []

        for field_name, rules in self.rules.items():
            # 检查字段是否存在于记录中
            if field_name not in record or pd.isna(record[field_name]):
                continue

            # 应用每个验证规则
            for rule_func, error_message in rules:
                try:
                    if not rule_func(record[field_name]):
                        errors.append(f"字段 '{field_name}': {error_message}")
                except Exception as e:
                    errors.append(f"字段 '{field_name}' 验证出错: {str(e)}")

        return errors

    def validate_dataframe(self, dataframe):
        """验证DataFrame中的所有记录

        Args:
            dataframe: 要验证的DataFrame

        Returns:
            dict: 包含记录索引和验证错误的字典
        """
        validation_results = {}

        for idx, row in dataframe.iterrows():
            errors = self.validate_record(row)
            if errors:
                validation_results[idx] = errors

        return validation_results

    def generate_validation_report(self, validation_results, output_file=None):
        """生成验证报告

        Args:
            validation_results: 验证结果字典
            output_file: 输出文件路径，如果为None则返回字符串

        Returns:
            str: 验证报告内容
        """
        report = []
        report.append("数据验证报告")
        report.append("=" * 50)

        total_errors = 0

        for idx, errors in validation_results.items():
            total_errors += len(errors)
            report.append(f"\n记录索引 {idx}:")
            for error in errors:
                report.append(f"  - {error}")

        report.append("=" * 50)
        report.append(f"总错误数: {total_errors}")
        report.append(f"有错误的记录数: {len(validation_results)}")

        report_content = "\n".join(report)

        # 如果指定了输出文件，将报告写入文件
        if output_file:
            try:
                with open(output_file, "w", encoding="utf-8") as f:
                    f.write(report_content)
            except Exception as e:
                print(f"写入验证报告失败: {str(e)}")

        return report_content
